package com.navinfo.platform.trip.analysis.flink.window;

import com.navinfo.platform.trip.analysis.flink.config.SystemConfig;
import com.navinfo.platform.trip.common.enums.SystemTypeEnum;
import com.navinfo.platform.trip.analysis.pojo.TripJudgeState;
import com.navinfo.platform.trip.analysis.service.ServiceFactory;
import com.navinfo.platform.trip.analysis.service.TripJudgeService;
import com.navinfo.platform.trip.common.enums.ProtocolEnum;
import com.navinfo.platform.trip.common.pojo.PointProtocol;
import com.navinfo.platform.trip.common.util.CommonUtils;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Optional;

/**
 * 行程切分触发器，一共有两种触发窗口计算的条件
 * 情况一： 行程正常结束，等待tripWindowWait后在onProcessingTime方法中触发
 * 情况二： 行程强制结束，最后一个点超过TripJudgeService.FORCE_END_DURATION还没有数据达到，在onProcessingTime方法中触发
 * @author: web
 * @data: 2020-04-07
 **/
public class TripWindowTrigger extends Trigger<PointProtocol, GlobalWindow> {
    private static final Logger logger = LoggerFactory.getLogger(TripWindowTrigger.class);

    /**
     * 最小速度，单位km/h
     * 最大速度，单位km/h
     */
    private int speedMin,speedMax;
    /**
     * 引擎最小转速
     * 引擎最大转速
     **/
    private int rotationMin,rotationMax;

    /**
     * 行程窗口达到触发条件后，窗口等待触发计算的时长<br/>
     * 单位：s，默认：60s
     */
    private int tripWindowWait;

    /**
     * 行程切分服务
     */
    private TripJudgeService tripJudgeService;

    /**
     * 行程状态判断的缓存数据<br/>
     * 由于窗口等待tripWindowWait秒，所有可能存在多个行程状态判断的缓存
     */
    private final ListStateDescriptor<TripJudgeState> tripListDesc = new ListStateDescriptor<>("trip_list_desc", PojoTypeInfo.of(TripJudgeState.class));

    /**
     * 强制结束行程时间点的缓存数据
     */
    private final ValueStateDescriptor<Long> forceFireDesc = new ValueStateDescriptor<>("force_fire_desc", Long.class);

    public TripWindowTrigger(ParameterTool jobConfig) {
        rotationMin = jobConfig.getInt("rotation_min", 0);
        rotationMax = jobConfig.getInt("rotation_max", 10000);
        speedMin = jobConfig.getInt("speed_min", 0);
        speedMax = jobConfig.getInt("speed_max", 200);
        tripWindowWait = jobConfig.getInt(SystemConfig.TRIP_WINDOW_WAIT_KEY, 60);

        tripJudgeService = ServiceFactory.createTripJudgeService(SystemTypeEnum.valueOf(jobConfig.getInt(SystemConfig.SYSTEM_TYPE_KEY)), jobConfig);
    }

    /**
     * 处理通过Union连接的0F37和0200两个流
     * @param element   传入元素
     * @param timestamp 事件时间，Source指定的事件时间提取器
     * @param window    窗口对象
     * @param ctx       上下文
     * @return 是否触发
     */
    @Override
    public TriggerResult onElement(PointProtocol element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {
        TriggerResult result = TriggerResult.CONTINUE;

        //数据合法性校验
        if(!checkData(element)){
            return result;
        }

        //删除上一个强制结束行程的定时器
        ValueState<Long> forceFireStore = ctx.getPartitionedState(forceFireDesc);
        Optional.ofNullable(forceFireStore.value()).ifPresent(ctx::deleteProcessingTimeTimer);

        //初始化行程状态判断的缓存数据
        ListState<TripJudgeState> tripListStore = ctx.getPartitionedState(tripListDesc);
        if(null == tripListStore.get()){
            tripListStore.add(new TripJudgeState());
        }
        
        //获取当前行程状态判断的缓存数据
        List<TripJudgeState> tripList = CommonUtils.toList(tripListStore.get().iterator());
        TripJudgeState firstTripState = tripList.get(0);
        TripJudgeState currTripState = tripList.get(tripList.size() - 1);
        if(currTripState.isFire()){
            currTripState = new TripJudgeState();
            tripList.add(currTripState);
        }

        //当前时间
        long currGpsTime = element.getGpsTime();

        //密集数据
        if(ProtocolEnum.REALTIME.equals(element.getProtocol())) {
            //当第一个行程的状态数据达到触发状态，并且0F37时间已大于第一个行程结束点，触发窗口计算
//            if (firstTripState.isFire() && null!=firstTripState.getFireTime() && currGpsTime>=firstTripState.getFireTime()) {
//                //移除当前行程状态判断的缓存数据
//                tripList.remove(0);
//                //触发窗口计算
//                result = TriggerResult.FIRE;
//            }
        }

        //位置数据
        else if(ProtocolEnum.LOCATION.equals(element.getProtocol())){
            switch (tripJudgeService.tripStatus(currTripState, element)) {
                case TRIP_START:
                case TRIP_IN:
                    currTripState.addCount0200(1);
                    Long lastGpsTime = currTripState.getLastGpsTime();
                    currTripState.setLowRotationDuration(null==lastGpsTime ? 0 : (currTripState.getLastRotation()<=500 && element.getRotation()<=500 ? currTripState.getLowRotationDuration()+(element.getGpsTime()-lastGpsTime) : 0)) ;
                    currTripState.setLastGpsTime(currGpsTime);
                    currTripState.setLastRotation(element.getRotation());
                    break;
                case TRIP_END:
                case TRIP_FORCE_END:
                    //设置行程结束数据，保证触发时间为奇数
                    long fireTime = System.currentTimeMillis() + tripWindowWait*1000;
                    fireTime = (fireTime & 1)==1 ? fireTime : fireTime+1;
                    currTripState.setFireTime(fireTime);
                    currTripState.setFire(true);

                    //增加延迟触发窗口计算的定时器
                    ctx.registerProcessingTimeTimer(currTripState.getFireTime());
                    break;
                default:
                    break;
            }
        }

        //增加强制结束行程的定时器
        if(result.equals(TriggerResult.CONTINUE)){
            //保证强制触发时间为偶数;
            long forceFireTime = System.currentTimeMillis() + TripJudgeService.FORCE_END_DURATION;
            forceFireTime = (forceFireTime & 1)==1 ? forceFireTime+1 : forceFireTime;
            forceFireStore.update(forceFireTime);
            ctx.registerProcessingTimeTimer(forceFireStore.value());
        }

        //更新行程状态判断的缓存数据
        tripListStore.update(tripList);
        return result;
    }

    /**
     * 延迟触发窗口计算的定时器
     * 强制结束行程的定时器
     * @param time 定时器的启动时间
     * @param window 当前窗口
     * @param ctx    上下文
     * @return       触发结果
     */
    @Override
    public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
        //延迟触发窗口计算的处理
        if((time & 1) == 1){
            ListState<TripJudgeState> tripListStore = ctx.getPartitionedState(tripListDesc);
            if(null != tripListStore.get()){
                List<TripJudgeState> tripList = CommonUtils.toList(tripListStore.get().iterator());
                if(tripList.size() > 0){
                    if(tripList.get(0).isFire()){
                        //移除当前行程状态判断的缓存数据
                        tripList.remove(0);
                        tripListStore.update(tripList);

                        //触发窗口计算
                        return TriggerResult.FIRE;
                    }
                }
            }
        }
        //强制结束行程的处理
        else{
            ValueState<Long> forceFireStore = ctx.getPartitionedState(forceFireDesc);
            Long forceFireTime = forceFireStore.value();
            if(null!=forceFireTime && time>=forceFireTime){
                forceFireStore.update(null);
                return TriggerResult.FIRE;
            }

        }

        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(GlobalWindow window, TriggerContext ctx) {
    }

    /**
     * 数据合法性校验
     * @param point 公关轨迹点
     * @return 检查结果
     */
    private boolean checkData(PointProtocol point) {
        //本次gps时间大于当前时间一个小时以上，不予处理
        if (point.getGpsTime() > System.currentTimeMillis()+3600000) {
            logger.info("===本次gps时间（{}）大于当前时间一个小时以上，不予处理! 通信号：{} 接收时间：{} 协议：{} ===", point.getGpsTime(), point.getTid(), point.getReceiveDate(), point.getProtocol());
            return false;
        }

        //发动机转速不在合理区间内
        int rotation = point.getRotation();
        if (rotation<rotationMin || rotation>rotationMax) {
            logger.info("===发动机转速不在合理区间内，不予处理!（转速：" + rotation + "rpm） 通信号：{} gps时间：{} 接收时间：{} 协议：{} ===", point.getTid(), point.getGpsTime(), point.getReceiveDate(), point.getProtocol());
            return false;

        }

        //速度不在合理区间内
        double speedKmh = point.getSpeedKmh();
        if (speedKmh<speedMin || speedKmh>speedMax) {
            logger.info("===速度不在合理区间内，不予处理!（速度：" + speedKmh + "km/h） 通信号：{} gps时间：{} 接收时间：{} 协议：{} ===",  point.getTid(), point.getGpsTime(), point.getReceiveDate(), point.getProtocol());
            return false;
        }

        return true;
    }

}
